package org.jeecg.modules.common.config.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.scheduling.annotation.Async;

@Configuration
@EnableConfigurationProperties(value = {MqttProperties.class})
@Slf4j
public class MqttConfig {

    public static final String OUTBOUND_CHANNEL = "mqttOutboundChannel";
    public static final String INPUT_CHANNEL = "mqttInputChannel";
    @Autowired
    MqttReceiveHandle mqttReceiveHandle;

    /**
     * 构造一个默认的mqtt客户端操作bean
     *
     * @return
     */
    @Bean
    public MqttPahoClientFactory clientFactory(MqttProperties properties) {
        final MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(properties.getUris());
        // 如果设置为true则接收不到掉线的消息，改为false则可以收到掉线的消息，
        options.setCleanSession(false);
        options.setUserName(properties.getUserName());
        options.setPassword(properties.getPassword().toCharArray());
        final DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean(value = OUTBOUND_CHANNEL)
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = OUTBOUND_CHANNEL)
    public MessageHandler mqttOutbound(MqttPahoClientFactory factory, MqttProperties properties) {
        final MqttPahoMessageHandler handler = new MqttPahoMessageHandler(properties.getClientId() + "_outbound", factory);
        handler.setDefaultQos(1);
        handler.setDefaultRetained(false);
        handler.setDefaultTopic(properties.getDefaultTopic());
        handler.setAsync(false);
        handler.setAsyncEvents(false);
        return handler;
    }

    /**
     * MQTT消息接收处理
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT接收消息通道适配器
     * @param factory
     * @param properties
     * @return
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory, MqttProperties properties){
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        properties.getClientId() + "_inbound", factory, properties.getSubscribeTopics());
        return adapter;
    }
    /**
     * 配置client,监听的topic
     * @return
     */
    @Bean
    public MessageProducer inbound(MqttPahoMessageDrivenChannelAdapter adapter, MessageChannel mqttInputChannel) {
        adapter.setCompletionTimeout(3000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel);
        return adapter;
    }

    /**
     * 通过通道获取数据
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public MessageHandler handler() {
        return message -> {
            mqttReceiveHandle.handle(message);
        };
    }

}
